package com.primeton.viewtest;

import java.util.Calendar;
import java.util.GregorianCalendar;
import java.util.Iterator;

import com.espertech.esper.client.EventBean;
import com.espertech.esper.client.EventType;
import com.espertech.esper.core.context.util.AgentInstanceViewFactoryChainContext;
import com.espertech.esper.core.service.EPStatementHandleCallback;
import com.espertech.esper.core.service.ExtensionServicesContext;
import com.espertech.esper.epl.expression.ExprNode;
import com.espertech.esper.event.EventAdapterService;
import com.espertech.esper.schedule.ScheduleHandleCallback;
import com.espertech.esper.schedule.ScheduleSlot;
import com.espertech.esper.view.CloneableView;
import com.espertech.esper.view.View;
import com.espertech.esper.view.ViewSupport;

public class FickView extends ViewSupport implements CloneableView {
	private final static int														LATE_EVENT_SLACK_SECONDS	= 5;

	private final AgentInstanceViewFactoryChainContext	agentInstanceViewFactoryContext;

	private final ScheduleSlot													scheduleSlot;

	private final ExprNode															timestampExpression;

	private final ExprNode															valueExpression;

	private final EventBean[]														eventsPerStream						= new EventBean[1];

	private EPStatementHandleCallback										handle;

	private Long																				cutoffTimestampMinute;

	private Long																				currentTimestampMinute;

	private Double																			first;

	private Double																			last;

	private Double																			max;

	private Double																			min;

	private EventBean																		lastEvent;

	public FickView(AgentInstanceViewFactoryChainContext agentInstanceViewFactoryContext, ExprNode timestampExpression,
			ExprNode valueExpression) {
		this.agentInstanceViewFactoryContext = agentInstanceViewFactoryContext;
		this.timestampExpression = timestampExpression;
		this.valueExpression = valueExpression;
		this.scheduleSlot = agentInstanceViewFactoryContext.getStatementContext().getScheduleBucket().allocateSlot();
	}

	@Override
	public void update(EventBean[] newData, EventBean[] oldData) {
		System.out.println("FickView  (1) update");
		if (newData == null) {
			return;
		}

		for (EventBean theEvent : newData) {
			eventsPerStream[0] = theEvent;
			Long timestamp = (Long) timestampExpression.getExprEvaluator().evaluate(eventsPerStream, true,
					agentInstanceViewFactoryContext);
			Long timestampMinute = removeSeconds(timestamp);
			double value = (Double) valueExpression.getExprEvaluator().evaluate(eventsPerStream, true,
					agentInstanceViewFactoryContext);

			// test if this minute has already been published, the event is too late
			if ((cutoffTimestampMinute != null) && (timestampMinute <= cutoffTimestampMinute)) {
				continue;
			}

			// if the same minute, aggregate
			if (timestampMinute.equals(currentTimestampMinute)) {
				applyValue(value);
			}
			// first time we see an event for this minute
			else {
				// there is data to post
				if (currentTimestampMinute != null) {
					postData();
				}

				currentTimestampMinute = timestampMinute;
				applyValue(value);

				// schedule a callback to fire in case no more events arrive
				scheduleCallback();
			}
		}
	}

	@Override
	public EventType getEventType() {
		System.out.println("FickView  (2)   getEventType");
		return getEventType(agentInstanceViewFactoryContext.getStatementContext().getEventAdapterService());
	}

	@Override
	public Iterator<EventBean> iterator() {
		throw new UnsupportedOperationException("Not supported");
	}

	@Override
	public View cloneView() {
		return new FickView(agentInstanceViewFactoryContext, timestampExpression, valueExpression);
	}

	private void applyValue(double value) {
		if (first == null) {
			first = value;
		}
		last = value;
		if (min == null) {
			min = value;
		} else if (min.compareTo(value) > 0) {
			min = value;
		}
		if (max == null) {
			max = value;
		} else if (max.compareTo(value) < 0) {
			max = value;
		}
	}

	protected static EventType getEventType(EventAdapterService eventAdapterService) {
		System.out.println("FickView  (3) getEventType customer!!");
		return eventAdapterService.addBeanType(FickValue.class.getName(), FickValue.class, false, false, false);
	}

	private static long removeSeconds(long timestamp) {
		Calendar cal = GregorianCalendar.getInstance();
		cal.setTimeInMillis(timestamp);
		cal.set(Calendar.SECOND, 0);
		cal.set(Calendar.MILLISECOND, 0);
		return cal.getTimeInMillis();
	}

	private void scheduleCallback() {
		if (handle != null) {
			// remove old schedule
			agentInstanceViewFactoryContext.getStatementContext().getSchedulingService().remove(handle, scheduleSlot);
			handle = null;
		}

		long currentTime = agentInstanceViewFactoryContext.getStatementContext().getSchedulingService().getTime();
		long currentRemoveSeconds = removeSeconds(currentTime);
		long targetTime = currentRemoveSeconds + (60 + LATE_EVENT_SLACK_SECONDS) * 1000; // leave
																																											// some
																																											// seconds
																																											// for
																																											// late
																																											// comers
		long scheduleAfterMSec = targetTime - currentTime;

		ScheduleHandleCallback callback = new ScheduleHandleCallback() {
			@Override
			public void scheduledTrigger(ExtensionServicesContext extensionServicesContext) {
				handle = null; // clear out schedule handle
				FickView.this.postData();
			}
		};

		handle = new EPStatementHandleCallback(agentInstanceViewFactoryContext.getEpStatementAgentInstanceHandle(),
				callback);
		agentInstanceViewFactoryContext.getStatementContext().getSchedulingService()
				.add(scheduleAfterMSec, handle, scheduleSlot);
	}

	private void postData() {
		FickValue fickValue = new FickValue(currentTimestampMinute, first, last, max, min);
		EventBean outgoing = agentInstanceViewFactoryContext.getStatementContext().getEventAdapterService()
				.adapterForBean(fickValue);
		if (lastEvent == null) {
			this.updateChildren(new EventBean[] { outgoing }, null);
		} else {
			this.updateChildren(new EventBean[] { outgoing }, new EventBean[] { lastEvent });
		}
		lastEvent = outgoing;

		cutoffTimestampMinute = currentTimestampMinute;
		first = null;
		last = null;
		max = null;
		min = null;
		currentTimestampMinute = null;
	}
}
